package io.gitee.kenewstar.dbmq.data;

import cn.hutool.core.collection.CollUtil;
import io.gitee.kenewstar.dbmq.ConsumerRegisterInfo;
import io.gitee.kenewstar.dbmq.ConsumerTask;
import io.gitee.kenewstar.dbmq.Message;
import io.gitee.kenewstar.dbmq.util.Const;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @author kenewstar
 */
public interface DbDataHandler {

    Integer COUNT = 100;

    /**
     * 重新分配消息数据
     *
     * @param taskList 消费者任务列表
     */
    default void reAssignMessage(List<ConsumerTask> taskList) {
        if (CollUtil.isEmpty(taskList)) {
            return;
        }

        final List<String> topics = taskList.stream().map(ConsumerTask::getTopic).collect(Collectors.toList());
        final List<ConsumerRegisterInfo> list = this.getRegisterInfo(topics);
        final Map<String, List<ConsumerRegisterInfo>> topicGroup = list.stream().collect(Collectors.groupingBy(ConsumerRegisterInfo::getTopic));

        for (ConsumerTask task : taskList) {
            final List<ConsumerRegisterInfo> infoList = topicGroup.get(task.getTopic());
            if (CollUtil.isEmpty(infoList)) {
                return;
            }

            List<Integer> tags = new ArrayList<>(infoList.size());
            for (int i = 0; i < infoList.size(); i++) {
                if (Objects.equals(infoList.get(i).getStatus(), Const.Status.UP)) {
                    tags.add(i + 1);
                }
            }

            if (CollUtil.isEmpty(tags)) {
                return;
            }
            // 获取最后一个消费者数据，判断是否为下线状态，如果是，则尝试获取
            final ConsumerRegisterInfo lastData = infoList.get(infoList.size() - 1);
            if (this.tryGetConsumerRegister(lastData)) {
                try {
                    // 获取成功
                    Integer tag = infoList.size();
                    final List<Message> reAssignMessage = this.getReAssignMessage(task.getTopic(), tag);
                    if (CollUtil.isNotEmpty(reAssignMessage)) {
                        this.batchUpdateMessage(reAssignMessage, tags, task.getTopic());
                    } else {
                        this.deleteConsumerRegister(lastData);
                    }

                } finally {
                    // 释放lock
                    this.releaseConsumerRegister(lastData);
                }
            }
        }
    }

    /**
     * 获取数据库类型
     *
     * @return type
     */
    String getType();

    /**
     * 初始化消费者注册表
     */
    void initConsumerRegister();

    /**
     * 初始化消费者队列表
     *
     * @param taskList 消费者任务列表
     */
    void initConsumerQueue(List<ConsumerTask> taskList);

    /**
     * 拉取所有消费者注册信息
     * @param topics 主题列表
     * @return 返回数据
     */
    List<ConsumerRegisterInfo> pullRegisterInfo(List<String> topics);

    /**
     * 创建消费者数据
     *
     * @param task 消费者任务
     * @return 返回数据
     */
    ConsumerRegisterInfo createConsumer(ConsumerTask task);

    /**
     * 更新消费者数据
     *
     * @param task 消费者任务
     * @param registerInfo 过期注册信息
     */
    void updateConsumer(ConsumerTask task, List<ConsumerRegisterInfo> registerInfo);

    /**
     * 消费者注册信息下线
     *
     * @param id 注册信息表ID
     */
    void registerDown(Long id);

    /**
     * 消费者注册信息下线
     * @param list 消费者信息列表
     */
    void registerDown(List<ConsumerRegisterInfo> list);

    /**
     * 消费者注册信息发送心跳
     *
     * @param ids 注册信息表ID
     */
    void sendHeartbeat(List<Long> ids);

    /**
     * 获取消息数据
     *
     * @param task 消费者任务
     * @return 消息数据
     */
    List<Message> getMessage(ConsumerTask task);

    /**
     * 删除消息
     *
     * @param topic 主题
     * @param ids 消息ID
     */
    void deleteMessage(String topic, List<Long> ids);

    /**
     * 发送消息数据
     *
     * @param topic 主题
     * @param messageBodyList 消息内容
     * @param tags 消息标签
     * @return rows 影响行数
     */
    int sendMessage(String topic, List<String> messageBodyList, List<Integer> tags);

    List<ConsumerRegisterInfo> getRegisterInfo(List<String> topics);

    boolean tryGetConsumerRegister(ConsumerRegisterInfo registerInfo);

    void releaseConsumerRegister(ConsumerRegisterInfo registerInfo);

    void deleteConsumerRegister(ConsumerRegisterInfo registerInfo);

    List<Message> getReAssignMessage(String topic, Integer tag);

    void batchUpdateMessage(List<Message> messageList, List<Integer> tags, String topic);

}